package david.java.flink.join.dataSource;

import david.java.flink.join.entity.Tx;
import lombok.SneakyThrows;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;

import java.io.*;

/**
 * @Description:
 * @Author: David
 * @Date: Create in 10:21 上午 2021/11/8
 */
public class TxDataSource implements SourceFunction<Tx> {

    private volatile boolean isRunning = true;
    private static BufferedReader br;
    private static Integer SLEEP = 1000;
    private static Integer MAX_OUT_OF_ORDERNESS = 10 * 1000;


    public TxDataSource() throws FileNotFoundException {
        br = new BufferedReader(new FileReader("src/main/resources/TxData"));
    }

    @Override
    public void run(SourceContext sourceContext) throws Exception {
        String line = null;
        while (null != (line = br.readLine())) {
            String[] split = line.split(",");
            Tx tx = new Tx(Integer.valueOf(split[0]), split[1], Double.valueOf(split[2]));
            long timestamp = Long.valueOf(split[3]);
            sourceContext.collect(tx);
            sourceContext.collectWithTimestamp(tx, timestamp);
            sourceContext.emitWatermark(new Watermark(timestamp - MAX_OUT_OF_ORDERNESS));
            Thread.sleep(SLEEP);
        }
    }

    @SneakyThrows
    @Override
    public void cancel() {
        isRunning = !isRunning;
        br.close();
    }
}
